use axum::{
    extract::{
        ws::{Message, WebSocket, WebSocketUpgrade},
        Extension,
    },
    response::IntoResponse,
    Json,
};
use serde::{Deserialize,Serialize};
use serde_json::{from_str, json};
use sqlx::{PgPool, Pool, Postgres};
use chrono::NaiveDate;
use std::{collections::HashMap, sync::Arc, hash};
use futures::{sink::SinkExt, stream::StreamExt};

use crate::state::AppState;


// use crate::{middleware::{db::get_pool, sendcatach::SendCatch}};

// use crate::middleware::error::Error;

#[derive(Serialize,Deserialize,Clone,Debug)]
pub  struct  MessageItem {
    from_user_openid:String, 
    to_user_openid :String,   
    msg:String,
}
#[derive(Serialize,Deserialize,Clone,Debug)]
pub  struct  SendMessageItem {
    from_user_openid:String, 
    to_user_openid :String,   
    msg:String,
    create_time: NaiveDate ,
}
#[derive(Serialize,Deserialize,Clone,Debug)]
pub  struct  SendMessage {
    from_user_openid:String,
    to_user_openid :String,  
    // msg:String, 
}



pub async fn ws_handler(
    ws: WebSocketUpgrade,
    Extension(pool): Extension<PgPool>,
    Extension(state): Extension<Arc<AppState>>,
) -> impl IntoResponse {
    ws.on_upgrade(|socket| post_socket(socket, pool,state))
}
// mut 
async fn post_socket(mut socket: WebSocket,pool: Pool<Postgres>,state:Arc<AppState>){ 
    //
    let (mut sender, mut receiver) = socket.split();
    let mut username = String::new();
    let mut rxusername= String::new();
    let mut from_user_openid = String::new();
    let mut to_user_openid = String::new();
    //第一次发消息 opne 时发送的消息
    while let Some(Ok(message)) = receiver.next().await {
        if let Message::Text(name) = message {
            if let  Ok(send_option) =from_str::<SendMessage>(&name){
                let id =format!("{}-{}",send_option.from_user_openid,send_option.to_user_openid); //标记 
                rxusername= format!("{}-{}", send_option.to_user_openid,send_option.from_user_openid);//订阅标记
                to_user_openid= send_option.to_user_openid; 
                from_user_openid= send_option.from_user_openid;
                check_username(&state, &mut username, &id);
            }
            // If not empty we want to quit the loop else we want to quit function.
            if !username.is_empty() {
                break;
            } else {
                // Only send our client that username is taken.
                let _ = sender
                    .send(Message::Text(String::from("Username already taken.")))
                    .await;
                return;
            }
        }
    }

    //发消息前进行订阅
    let mut rx = state.tx.subscribe();

    //将合并消息发送给所有订阅者。
    let msg = format!("{}", username);
    let _ = state.tx.send(msg);
    let isname= username.clone();

    let mut send_task = tokio::spawn(async move {
        while let Ok(msg) = rx.recv().await {
            //接受数据并响应
            let msg_vec:Vec<&str> =  msg.split(':').collect();
     
            if msg_vec[0] == rxusername { //给订阅的客户发送消息
                match  msg_vec.len() {
                    2=>{
                        if sender.send(Message::Text( msg_vec[1].to_owned())).await.is_err() {
                            break;  
                        }
                    },
                    _=>{},
                }
            }
        }
    });
    
    let tx = state.tx.clone();
    let name = username.clone(); //订阅者

    let mut recv_task = tokio::spawn(async move {
        while let Some(Ok(Message::Text(text))) = receiver.next().await {
            let message=MessageItem{
                from_user_openid : format!("{}",from_user_openid),
                to_user_openid: format!("{}",to_user_openid),
                msg:text.clone(),
            };
            save_message(&message,&pool);
            let _ = tx.send(format!("{}:{}", name, text)); //发送数据
        }
    });

    // If any one of the tasks exit, abort the other.11
    tokio::select! {
        _ = (&mut send_task) => recv_task.abort(),
        _ = (&mut recv_task) => send_task.abort(),
    };

    let msg = format!("{}", username);
    let _ = state.tx.send(msg);

    state.send_catch.lock().unwrap().remove(&username); //关闭时清除

}

//查询username
fn check_username(state: &AppState, string: &mut String, name: &str) {
    let mut send_catch = state.send_catch.lock().unwrap();
    if !send_catch.contains(name) {
        send_catch.insert(name.to_owned());
        string.push_str(name);
    }
}

//保存输入信息
async fn save_message(message:&MessageItem,pool:&PgPool){
    let _=  sqlx::query!("INSERT INTO chatrooms (from_user_openid,to_user_openid,msg) VALUES($1,$2,$3)",
    message.from_user_openid,message.to_user_openid,message.msg).fetch_one(pool).await.is_err();

}
//获取信息
async fn get_message(pool:&PgPool )->Vec<SendMessageItem>{
    let res:Vec<SendMessageItem> = sqlx::query!("select * from chatrooms order by create_time limit 10").fetch_all(pool).await.unwrap()
        .into_iter().map(| x|SendMessageItem { from_user_openid: x.from_user_openid, to_user_openid: x.to_user_openid, msg:x.msg, create_time: x.create_time.unwrap() })
        .collect();
    return res ;  
}
// async fn get_message_last(pool:&PgPool )->SendMessageItem{
//     let res= sqlx::query!("select * from chatrooms limit 1 ").fetch_one(pool).await.unwrap();
//     let last = SendMessageItem{
//         from_user_openid: res.from_user_openid,
//         to_user_openid: res.to_user_openid,
//         msg: res.msg,
//         create_time: res.create_time.unwrap(),
//     };
//     return last ;  
// }